自定义损失函数和指标¶
Note
自定损失函数和指标是常见的需求，我们看看在tensorflow中如何完成它们。
函数定义¶
import tensorflow as tf
def huber_fn(y_true, y_pred):
"""Huber损失: 小时l2大时l1"""
error = y_true - y_pred
is_small_error = tf.abs(error) < 1
squared_loss = tf.square(error) / 2
linear_loss = tf.abs(error) - 0.5
# tf.where(cond, a, b): a if cond else b
return tf.where(is_small_error, squared_loss, linear_loss)
from tensorflow import keras
import utils
model = keras.models.load_model("my_housing_model/")
# 像其他损失函数一样使用
model.compile(loss=huber_fn, optimizer="nadam")
子类定义¶
通过创建 keras.losses.Loss 类的子类来定义。
class HuberLoss(keras.losses.Loss):
def __init__(self, threshold=1.0, **kwargs):
super().__init__(**kwargs)
self.threshold = threshold
def call(self, y_true, y_pred):
# 计算损失函数
error = y_true - y_pred
is_small_error = tf.abs(error) < self.threshold
squared_loss = tf.square(error) / 2
linear_loss = self.threshold * tf.abs(error) - self.threshold**2 / 2
return tf.where(is_small_error, squared_loss, linear_loss)
def get_config(self):
# 获得超参数
base_config = super().get_config()
return {**base_config, "threshold": self.threshold}
# 一样的使用
model.compile(loss=HuberLoss(2.0), optimizer="nadam")
指标¶
损失和指标不是一回事。
损失在梯度下降中用于训练模型，必须可微。
指标用于评估模型，必须更容易被解释。
但有时候，损失也可以作为指标，比如说 HuberLoss。
def create_huber(threshold=1.0):
"""带阈值的Huber损失"""
def huber_fn(y_true, y_pred):
error = y_true - y_pred
is_small_error = tf.abs(error) < threshold
squared_loss = tf.square(error) / 2
linear_loss = threshold * tf.abs(error) - threshold**2 / 2
return tf.where(is_small_error, squared_loss, linear_loss)
return huber_fn
class HuberMetric(keras.metrics.Metric):
"""Huber指标，继承自keras.metrics.Metric"""
def __init__(self, threshold=1.0, **kwargs):
# handles base args (e.g., dtype)
super().__init__(**kwargs)
self.threshold = threshold
# 计算损失的函数
self.huber_fn = create_huber(threshold)
# 损失总值和总次数
self.total = self.add_weight("total", initializer="zeros")
self.count = self.add_weight("count", initializer="zeros")
def update_state(self, y_true, y_pred, sample_weight=None):
# 给定一个批次的标签和预测值，如何更新变量
metric = self.huber_fn(y_true, y_pred)
self.total.assign_add(tf.reduce_sum(metric))
self.count.assign_add(tf.cast(tf.size(y_true), tf.float32))
def result(self):
# 计算并返回最终结果
return self.total / self.count
def get_config(self):
base_config = super().get_config()
return {**base_config, "threshold": self.threshold}
# 在编译时指定metrics
model.compile(loss=create_huber(2.0), optimizer="nadam", metrics=[HuberMetric(3.0)])